﻿using Misakai.Kafka;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading.Tasks;

namespace Reflexive.IO.Kafka
{
    public class KafkaConnectionFactory : IKafkaConnectionFactory
    {
        public IKafkaConnection Create(KafkaEndpoint endpoint, TimeSpan responseTimeoutMs, IKafkaLog log)
		{
            return new Reflexive.IO.Kafka.KafkaConnection(new Reflexive.IO.Kafka.KafkaTcpSocket(log, endpoint, 0), new TimeSpan?(responseTimeoutMs), log);
		}
		public KafkaEndpoint Resolve(Uri kafkaAddress, IKafkaLog log)
		{
            IPAddress firstAddress = Reflexive.IO.Kafka.KafkaConnectionFactory.GetFirstAddress(kafkaAddress.Host, log);
			IPEndPoint endpoint = new IPEndPoint(firstAddress, kafkaAddress.Port);
			return new KafkaEndpoint
			{
				ServeUri = kafkaAddress,
				Endpoint = endpoint
			};
		}
		private static IPAddress GetFirstAddress(string hostname, IKafkaLog log)
		{
			try
			{
				IPAddress[] hostAddresses = Dns.GetHostAddresses(hostname);
				if (hostAddresses.Length > 0)
				{
					Array.ForEach<IPAddress>(hostAddresses, delegate(IPAddress address)
					{
						log.DebugFormat("Found address {0} for {1}", new object[]
						{
							address,
							hostname
						});
					});
					IPAddress iPAddress = hostAddresses.FirstOrDefault((IPAddress item) => item.AddressFamily == AddressFamily.InterNetwork) ?? hostAddresses.First<IPAddress>();
					log.DebugFormat("Using address {0} for {1}", new object[]
					{
						iPAddress,
						hostname
					});
					return iPAddress;
				}
			}
			catch
			{
				throw new UnresolvedHostnameException("Could not resolve the following hostname: {0}", new object[]
				{
					hostname
				});
			}
			throw new UnresolvedHostnameException("Could not resolve the following hostname: {0}", new object[]
			{
				hostname
			});
		}
	}

}
